package com.loong.spark.scala.graphx

import org.apache.spark._
import org.apache.spark.graphx.{GraphLoader, GraphXUtils, PartitionStrategy}
import org.apache.spark.storage.StorageLevel

import scala.collection.mutable

/**
  * Created by 梁浩峰 on 2016/9/13 22:54.
  */
object Analytics extends Logging {
  def main(args: Array[String]): Unit = {
    if (args.length < 2) {
      System.err.println("Usage: Analytics <taskType> <file> --numEPart=<num_dage_partitions> [other options]")
      System.err.println("Supported 'taskType' as follows:")
      System.err.println(" pagerank Compute PageRank")
      System.err.println(" cc       Compute the connected components of vertices")
      System.err.println(" triangles Count the number of triangles")
      System.exit(1)
      Math
    }

    val taskType = args(0)
    val fname = args(1)
    val optionsList = args.drop(2).map { arg =>
      arg.dropWhile(_ == '-').split("=") match {
        case Array(opt, v) => (opt -> v)
        case _ => throw new IllegalArgumentException("Invalid argument: " + arg)
      }
    }

    val options = mutable.Map(optionsList: _*)

    val conf = new SparkConf().setAppName("AnalyticsApp").setMaster("local")
    GraphXUtils.registerKryoClasses(conf)

    val numEPart = options.remove("numEPart").map(_.toInt).getOrElse {
      println("Set the number of edge partitions using --numEPart.")
      sys.exit(1)
    }

    val partitionStrategy: Option[PartitionStrategy] = options.remove("partStrategy").map(PartitionStrategy.fromString(_))

    val edgeStorageLevel = options.remove("edgeStorageLevel").map(StorageLevel.formatted(_)).getOrElse(StorageLevel.MEMORY_ONLY)
    val vertexStorageLevel = options.remove("vertexStorageLevel").map(StorageLevel.formatted(_)).getOrElse(StorageLevel.MEMORY_ONLY)

    /*taskType mastch {
      case "pagerank" =>
        val tol = options.remove("tol").map(_.toFloat).getOrElse(0.001F)
        val outFname = options.remove("output").getOrElse("")
        val numIterOpt = options.remove("numIter").map(_.toInt)
        options.foreach {
          case (opt, _) => throw new IllegalArgumentException("Invalid option: " + opt)
        }
        println("=========================================")
        println("|            PageRank                   |")
        println("=========================================")

        val sc = new SparkContext(conf.setAppName("PageRank(" + fname + ")"))

        val unpartitionedGraph = GraphLoader.edgeListFile(sc, fname, numEdgePartitions = numEPart, edgeStorageLevel = edgeStorageLevel, vertexStorageLevel = vertexStorageLevel).cache()
        val graph = partitionStrategy.foldLeft(unpartitionedGraph)(_.partitionBy(_))
    }
*/
  }
}
